package com.cn.task;

import com.cn.Utils.RunLogQueue;
import com.cn.Utils.SpringUtil;
import com.cn.event.RunLogEvent;
import com.cn.service.RunLogService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

/**
 * Created by chenyuncong on 2018/8/6.
 */
public abstract class SaveConsumer extends Consumer {

    public static ConcurrentLinkedQueue rlQueue = new ConcurrentLinkedQueue();

    private ScheduledExecutorService scheduledThreadPool = null;

    private ExecutorService singleExecutor = null;

    private int num=0;

    private int saveNum=100;

    public SaveConsumer(){
        this.scheduledThreadPool = Executors.newScheduledThreadPool(1);
        this.singleExecutor = Executors.newSingleThreadExecutor();
    }

    public void init(){
        this.start();
        startScheduledTask(scheduledThreadPool);
    }

    private class Works implements Runnable{
        private ConcurrentLinkedQueue clQueue;
        private int saveNum;
        private boolean sign;
        public Works(ConcurrentLinkedQueue queue ,int saveNum,boolean sign) {
            this.clQueue=queue;
            this.saveNum=saveNum;
            this.sign=sign;
        }
        @Override
        public void run() {
            try{
                if(!clQueue.isEmpty()){
                    List list = new ArrayList();
                    for(int i =0; i <saveNum;i++){
                        if (!clQueue.isEmpty()) {
                            list.add(clQueue.poll());
                        } else {
                            break;
                        }
                    }
                    workTask(list,saveNum,sign);
                }
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }
    public abstract void workTask(List list ,int saveNum,boolean sign);

    private void  startScheduledTask(ScheduledExecutorService scheduledThreadPool){
        // 开始运行1秒，然后每一秒都运行一次
        scheduledThreadPool.scheduleAtFixedRate(new Works(rlQueue,saveNum,false),1, 1, TimeUnit.SECONDS);
    }

    @Override
    public void run() {

        while (true){
            try {
                rlQueue.offer(RunLogQueue.queueSave.take());
                if(num==100){  //每慢100条就提交一次
                    singleExecutor.submit(new Works(rlQueue,saveNum,true));
                    num=0;
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            num++;
        }

    }
}
